package com.lyf.scm.plan.domain.service.impl;

import com.lyf.scm.common.constants.BigDataEsConstants;
import com.lyf.scm.common.constants.CommonConstants;
import com.lyf.scm.common.enums.ChannelTypeEnum;
import com.lyf.scm.common.enums.SalesOtherTypeEnum;
import com.lyf.scm.plan.config.RedisConfig;
import com.lyf.scm.plan.domain.convertor.SalesPlanEsConvertor;
import com.lyf.scm.plan.domain.remote.item.dto.SkuPriceDTO;
import com.lyf.scm.plan.domain.service.RedisCacheService;
import com.lyf.scm.plan.domain.service.SalesPlanDirectlySynEsService;
import com.lyf.scm.plan.insfrastructure.db.dataobject.es.BigdataPredictlyEsEntity;
import com.lyf.scm.plan.insfrastructure.db.dataobject.es.SalesPlanEsEntity;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.data.elasticsearch.core.ScrolledPage;
import org.springframework.data.elasticsearch.core.query.IndexQuery;
import org.springframework.data.elasticsearch.core.query.IndexQueryBuilder;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.math.BigDecimal;
import java.time.LocalDate;
import java.time.format.DateTimeFormatter;
import java.time.temporal.WeekFields;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import static org.elasticsearch.index.query.QueryBuilders.*;

/**
 * @author Martin
 * @date 2020-10-26 15:53
 */
@Slf4j
@Service
public class SalesPlanDirectlySynEsServiceImpl implements SalesPlanDirectlySynEsService {

    private static final String SALES_PLAN_INDEX_NAME = "scm_plan_sales_requirement_";
    // 价格map，key是sku
   // private Map<String, BigDecimal> avgPriceMap = new HashMap<>();
    private Boolean local;
    private int count;
    @Resource
    private ElasticsearchTemplate elasticsearchTemplate;
    @Resource
    private SalesPlanEsConvertor salesPlanEsConvertor;
    @Resource
    private RedisCacheService redisCacheService;
    @Resource
    private RedisConfig redisConfig;

    @Override
    public void transData2SalesPlanEs(String specifiedDate) {
        try {
            // 初始化缓存价格map
           // avgPriceMap = new HashMap<>();
            boolean powerSwitch = redisConfig.setNX(BigDataEsConstants.POWER_SWITCH, "1", 60 * 60 * 6);
            if (!powerSwitch) {
                return;
            }
            local = true;
            long start = System.currentTimeMillis();
            // 先检查数据是否已经跑过了或者在跑
            String indexName = SALES_PLAN_INDEX_NAME + specifiedDate.replace("-", "");
            if (isExistsIndex(indexName)) {
                // 检查索引是否存在
                if (!elasticsearchTemplate.indexExists(BigDataEsConstants.INDEX_NAME_PLAN_SALES_PLAN)) {
                    elasticsearchTemplate.createIndex(SalesPlanEsEntity.class);
                }
                // 同步数据
                syncData(indexName,specifiedDate);
                long end = System.currentTimeMillis();
                log.warn("总耗时：{}ms，约{}s", (end - start), (end - start) / 1000);
            }
        } catch (Exception e) {
            log.error("同步直营预测数据异常:", e);
        }

    }

    @SneakyThrows
    private Boolean isExistsIndex(String indexName) {
        if (elasticsearchTemplate.indexExists(indexName)) {
            return Boolean.TRUE;
        }
        TimeUnit.MINUTES.sleep(1);
        log.warn("第{}次检查索引是否存在", count++);
        System.out.println("第" + count++ + "次检查索引是否存在");
        return isExistsIndex(indexName);
    }

    /**
     * 同步数据
     *
     * @param indexName
     * @param specifiedDate
     */
    private void syncData(String indexName, String specifiedDate) {
        final long SCROLL_TIMEOUT = 5 * 60 * 1000;
        int pageNum = 0;
        int pageSize = 1000;
        NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder()
                .withQuery(
                        boolQuery()
                                .must(matchQuery("node_type", "store"))
                )
                .withIndices(indexName)
                .withPageable(PageRequest.of(pageNum, pageSize));
        ScrolledPage<BigdataPredictlyEsEntity> scroll = (ScrolledPage<BigdataPredictlyEsEntity>) elasticsearchTemplate
                .startScroll(SCROLL_TIMEOUT, queryBuilder.build(), BigdataPredictlyEsEntity.class);
        log.warn("总共需处理数据:{}条", scroll.getTotalElements());
        while (scroll.hasContent()) {
            List<BigdataPredictlyEsEntity> content = scroll.getContent();
            this.convertAndSaveEs(content,specifiedDate);
            scroll = (ScrolledPage<BigdataPredictlyEsEntity>) elasticsearchTemplate.continueScroll(
                    scroll.getScrollId(), SCROLL_TIMEOUT, BigdataPredictlyEsEntity.class);
        }
    }

    /**
     * 转换并保存数据
     *
     * @param content
     * @param specifiedDate
     */
    private void convertAndSaveEs(List<BigdataPredictlyEsEntity> content, String specifiedDate) {
        List<SalesPlanEsEntity> list = new ArrayList<>();
        for (BigdataPredictlyEsEntity entity : content) {
            LocalDate tempeDate = LocalDate.parse(specifiedDate, DateTimeFormatter.ofPattern("yyyy-MM-dd"))
                    .with(WeekFields.of(Locale.FRANCE).dayOfWeek(), CommonConstants.ONE);
            // 解析13周数据（13列转13行）
            for (int i = 1; i < 14; i++) {
                SalesPlanEsEntity salesPlanEsEntity = completeData(entity, i, tempeDate);
                if (salesPlanEsEntity == null) {
                    continue;
                }
                list.add(salesPlanEsEntity);
                // 加一周
                tempeDate = tempeDate.plusWeeks(1);

            }
        }
        this.asyncSaveOrUpdate(list);
    }

    /**
     * 完善数据
     *
     * @param entity
     * @param week
     * @param startDate
     * @return
     */
    private SalesPlanEsEntity completeData(BigdataPredictlyEsEntity entity, int week, LocalDate startDate) {
        try {
            SalesPlanEsEntity salesPlanEsEntity = salesPlanEsConvertor.bigdata2SalesPlan(entity);
            Method method = BigdataPredictlyEsEntity.class.getMethod("getSales_week_f_" + week);
            Object v = method.invoke(entity);
            BigDecimal perSales = v != null ? new BigDecimal(v.toString()) : BigDecimal.ZERO;
            salesPlanEsEntity.setPredict_sales(perSales);
            salesPlanEsEntity.setStart_date(startDate);
            salesPlanEsEntity.setEnd_date(startDate.plusDays(6).toString());

            // 拼区域，310000/312000/312700
            salesPlanEsEntity.setArea(getJoinedArea(entity.getProvince_code(), entity.getCity_code(), entity.getDistrict_code(), entity.getNode_id()));
            // 渠道,固定直营门店
            salesPlanEsEntity.setChannel(ChannelTypeEnum.DIRECTLY_OPERATE.getCode());
            // 计划类型是直营预测17
            salesPlanEsEntity.setPlan_type(SalesOtherTypeEnum.PLAN_DIRECTLY.getType());
            salesPlanEsEntity.setLast_predict_sales(BigDecimal.ZERO);
            salesPlanEsEntity.setConfirm_l2(perSales);
            salesPlanEsEntity.setConfirm_l3(perSales);
            salesPlanEsEntity.setConfirm_l4(perSales);
            salesPlanEsEntity.setLast_confirm_l2(BigDecimal.ZERO);
            salesPlanEsEntity.setLast_confirm_l3(BigDecimal.ZERO);
            salesPlanEsEntity.setLast_confirm_l4(BigDecimal.ZERO);
            salesPlanEsEntity.setConfirm_status_l2(CommonConstants.ONE);
            salesPlanEsEntity.setConfirm_status_l3(CommonConstants.ONE);
            salesPlanEsEntity.setConfirm_status_l4(CommonConstants.ONE);
            salesPlanEsEntity.setConfirm_target_l2(perSales);
            salesPlanEsEntity.setConfirm_target_l3(perSales);
            salesPlanEsEntity.setConfirm_target_l4(perSales);
            //暂时不取价格，设为0
            salesPlanEsEntity.setAveragePrice(BigDecimal.ZERO);
            // id置为空，插入
            salesPlanEsEntity.setId(null);
            return salesPlanEsEntity;
        } catch (Exception e) {
            log.error("转换BigdataPredictlyEsEntity数据出错", e);
        }
        return null;
    }

    private String getJoinedArea(String province, String city, String country, String nodeId) {
        if (StringUtils.isEmpty(province)) {
            return null;
        }
        String result = province;
        if (StringUtils.isEmpty(city)) {
            return result;
        }
        result = result + "/" + city;
        if (StringUtils.isEmpty(country)) {
            return result;
        }
        result = result + "/" + country;
        if (StringUtils.isEmpty(nodeId)) {
            return result;
        }
        return result + "/" + nodeId;
    }

    /**
     * 异步写es
     *
     * @param list
     */
    private void asyncSaveOrUpdate(List<SalesPlanEsEntity> list) {
        if (!local) {
            return;
        }
        long start = System.currentTimeMillis();
        if (elasticsearchTemplate.indexExists(SalesPlanEsEntity.class)) {
            Map<String, SalesPlanEsEntity> existedData = getExistedData(list);
            this.merge(list, existedData);
        }
        // todo 设置单价
        // setAvgPrice(list);

        List<IndexQuery> indexQueryList = new ArrayList<>();
        for (SalesPlanEsEntity salesPlanEsEntity : list) {
            IndexQuery indexQuery = new IndexQueryBuilder().withObject(salesPlanEsEntity).build();
            indexQueryList.add(indexQuery);
        }
        elasticsearchTemplate.bulkIndex(indexQueryList);
        long end = System.currentTimeMillis();
        log.warn("保存{}条数据，耗时{}ms，约{}s", list.size(), (end - start), (end - start) / 1000);
    }

    private Map<String, SalesPlanEsEntity> getExistedData(List<SalesPlanEsEntity> list) {
        Map<String, SalesPlanEsEntity> map = new HashMap<>();
        Set<String> storeSet = new HashSet<>();
        Set<String> skuSet = new HashSet<>();
        for (SalesPlanEsEntity salesPlanEsEntity : list) {
            storeSet.add(salesPlanEsEntity.getNode_id());
            skuSet.add(salesPlanEsEntity.getSku_key());
        }
        BoolQueryBuilder boolQueryBuilder = boolQuery();
        // 用.keyword会快很大，精确匹配
        setMultiConditions(boolQueryBuilder, "node_id", new ArrayList<>(storeSet));
        setMultiConditions(boolQueryBuilder, "sku_key", new ArrayList<>(skuSet));
        //加日期条件
        String minDate = list.stream().map(SalesPlanEsEntity::getStart_date)
                .min(Comparator.naturalOrder()).orElse(LocalDate.now().toString());
        String endDate = list.stream().map(SalesPlanEsEntity::getStart_date)
                .max(Comparator.naturalOrder()).orElse(LocalDate.now().plusWeeks(13).toString());
        boolQueryBuilder.must(rangeQuery("start_date").gte(minDate).lte(endDate));
        NativeSearchQuery searchQuery = new NativeSearchQueryBuilder()
                .withQuery(boolQueryBuilder)
                .withPageable(PageRequest.of(0, 10000))
                .withIndices(BigDataEsConstants.INDEX_NAME_PLAN_SALES_PLAN)
                .build();
        List<SalesPlanEsEntity> existedData = elasticsearchTemplate.queryForList(searchQuery, SalesPlanEsEntity.class);
        if (existedData == null || existedData.size() == 0) {
            return map;
        }
        for (SalesPlanEsEntity entity : existedData) {
            map.put(getKey(entity), entity);
        }
        return map;
    }

    /**
     * 设置查询条件
     *
     * @param boolQueryBuilder
     * @param field
     * @param values
     */
    private void setMultiConditions(BoolQueryBuilder boolQueryBuilder, String field, List<String> values) {
        if (values == null || values.size() == 0) {
            return;
        }
        BoolQueryBuilder subBoolQueryBuilder = boolQuery();
        values.forEach(v -> subBoolQueryBuilder.should(QueryBuilders.matchQuery(field, v)));
        boolQueryBuilder.must(subBoolQueryBuilder);
    }

    private String getKey(SalesPlanEsEntity entity) {
        return entity.getNode_id() + "|" + entity.getSku_key() + "|" + entity.getStart_date();
    }

    private void merge(List<SalesPlanEsEntity> list, Map<String, SalesPlanEsEntity> map) {
        for (SalesPlanEsEntity entity : list) {
            //暂时不取价格，设为0
            entity.setAveragePrice(BigDecimal.ZERO);
            SalesPlanEsEntity salesPlanEsEntity = map.get(getKey(entity));
            if (salesPlanEsEntity == null) {
                continue;
            }
            entity.setId(salesPlanEsEntity.getId());
            entity.setLast_predict_sales(salesPlanEsEntity.getPredict_sales() == null ? BigDecimal.ZERO : salesPlanEsEntity.getPredict_sales());
            entity.setConfirm_l2(salesPlanEsEntity.getPredict_sales() == null ? BigDecimal.ZERO : salesPlanEsEntity.getPredict_sales());
            entity.setConfirm_l3(salesPlanEsEntity.getPredict_sales() == null ? BigDecimal.ZERO : salesPlanEsEntity.getPredict_sales());
            entity.setConfirm_l4(salesPlanEsEntity.getPredict_sales() == null ? BigDecimal.ZERO : salesPlanEsEntity.getPredict_sales());
            entity.setLast_confirm_l2(salesPlanEsEntity.getConfirm_l2() == null ? BigDecimal.ZERO : salesPlanEsEntity.getConfirm_l2());
            entity.setLast_confirm_l3(salesPlanEsEntity.getConfirm_l3() == null ? BigDecimal.ZERO : salesPlanEsEntity.getConfirm_l3());
            entity.setLast_confirm_l4(salesPlanEsEntity.getConfirm_l4() == null ? BigDecimal.ZERO : salesPlanEsEntity.getConfirm_l4());
            entity.setConfirm_status_l2(CommonConstants.ONE);
            entity.setConfirm_status_l3(CommonConstants.ONE);
            entity.setConfirm_status_l4(CommonConstants.ONE);
            if (salesPlanEsEntity.getSs() != null) {
                entity.setSs(salesPlanEsEntity.getSs());
            }
        }
    }

    /**
     * 设置价格
     *
     * @param list
     */
    private void setAvgPrice(List<SalesPlanEsEntity> list) {
        List<SkuPriceDTO> skuPriceDTOList = list.stream()
                .map(e -> {
                    SkuPriceDTO skuPriceDTO = new SkuPriceDTO();
                    skuPriceDTO.setSkuCode(e.getSku_key());
                    skuPriceDTO.setSaleUnitCode(e.getUnit());
                    return skuPriceDTO;
                })
                .collect(Collectors.toList());
        Map<String, BigDecimal> skuAveragePrice = redisCacheService.getSkuAveragePrice(skuPriceDTOList);
        list.forEach(dto -> {
            BigDecimal price = skuAveragePrice.get(dto.getSku_key());
            dto.setAveragePrice(price == null ? BigDecimal.ZERO : price);
        });
    }

    /**
     * 停止任务
     *
     * @return
     */
    @Override
    public void stopTask() {
        redisConfig.remove(BigDataEsConstants.POWER_SWITCH);
        local = false;
    }

}
